package com.bw.gmall.realtime.dws.app;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeTrademarkCategoryUserRefundBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.dws.function.AsyncDimFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.app
 * @className: TradeTrademarkCategoryUserRefundBean
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/7 10:29
 * @version: 1.0
 */

public class DwsTradeTrademarkCategoryUserRefundWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeTrademarkCategoryUserRefundWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_REFUND, Constant.DWS_TRADE_SKU_ORDER_USER_TAG_WINDOW, 4, 10031);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
//         dataStreamSource.print();
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getEtl = getGetEtl(dataStreamSource);
//        getEtl.print();
        /**
         * 3）补充与分组相关的维度信息
         * （1）关联sku_info表
         *
         *   todo 未完成'
         *
         * （2）获取tm_id，category3_id。
         */
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getSkuInfo = getSkuInfo(getEtl);
//         getSkuInfo.print();
        //设置水位线
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> wmStream = getWmStream(getSkuInfo);
//         wmStream.print();

        wmStream.keyBy(new KeySelector<TradeTrademarkCategoryUserRefundBean, Tuple3<String,String,String>>() {
            @Override
            public Tuple3<String, String, String> getKey(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean) throws Exception {
                return Tuple3.of(tradeTrademarkCategoryUserRefundBean.getUserId(),tradeTrademarkCategoryUserRefundBean.getTrademarkName(),tradeTrademarkCategoryUserRefundBean.getCategory3Id());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(2)))
                .reduce(new ReduceFunction<TradeTrademarkCategoryUserRefundBean>() {
                    @Override
                    public TradeTrademarkCategoryUserRefundBean reduce(TradeTrademarkCategoryUserRefundBean t1, TradeTrademarkCategoryUserRefundBean t2) throws Exception {
                        t1.getOrderIdSet().addAll(t2.getOrderIdSet());
                        return t1;
                    }
                }, new ProcessWindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, Tuple3<String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple3<String, String, String> stringStringStringTuple3, ProcessWindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, Tuple3<String, String, String>, TimeWindow>.Context context, Iterable<TradeTrademarkCategoryUserRefundBean> iterable, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {
                        TradeTrademarkCategoryUserRefundBean bean = iterable.iterator().next();

                        bean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(context.window().getStart()));
                        bean.setRefundCount((long) bean.getOrderIdSet().size());
                        collector.collect(bean);
                    }
                }).print();

}

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getWmStream(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getSkuInfo) {
        return getSkuInfo.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeTrademarkCategoryUserRefundBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<TradeTrademarkCategoryUserRefundBean>() {
            @Override
            public long extractTimestamp(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean, long l) {
                return tradeTrademarkCategoryUserRefundBean.getTs();
            }
        }));
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getSkuInfo(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getEtl) {
        return AsyncDataStream.unorderedWait(getEtl, new AsyncDimFunction<TradeTrademarkCategoryUserRefundBean>("dim_sku_info") {
            @Override
            public void getResult(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean, JSONObject jsonObject) {
                tradeTrademarkCategoryUserRefundBean.setTrademarkId(jsonObject.getString("tm_id"));
                tradeTrademarkCategoryUserRefundBean.setCategory3Id(jsonObject.getString("category3_id"));
            }

            @Override
            public String getDimId(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean) {
                return tradeTrademarkCategoryUserRefundBean.getSkuId();
            }
        }, 5000, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getGetEtl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, TradeTrademarkCategoryUserRefundBean>() {
            @Override
            public void flatMap(String s, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {
                if (s != null) {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    TradeTrademarkCategoryUserRefundBean build = TradeTrademarkCategoryUserRefundBean.builder()
                            .orderIdSet(new HashSet<>(Collections.singleton(jsonObject.getString("order_id"))))
                            .skuId(jsonObject.getString("sku_id"))
                            .userId(jsonObject.getString("user_id"))
                            .ts(jsonObject.getLong("ts") * 1000)
                            .build();
                    collector.collect(build);
                }
            }

        });
    }
}
